package com.zhaosc.rx.example;

import org.apache.log4j.Logger;
import org.junit.Test;

import com.zhaosc.rx.utils.SleepUtils;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;

/**
 * 在异步场景下，被观察者发送事件速度远快于观察者处理的速度 ， 从而导致下游的 buffer 溢出 ， 这种现 象叫作背压 。
 * 
 * 首先，背压必须是在异步的场景下才会出现 ，即 被观察者和观察者处于不同的线程中 。 其次， Rx.Java是基于 Push模型的。对于 Pull模型而言，
 * 当消费者请求数据的时候，如果 生产者比较慢 ，则消 费者会阻塞等待。 如果生产者比较快， 则 生产者会等待消费者处理完后再 生产新的数据， 所以不会出
 * 现背压的情况 。 然而在 Rx.Java 中 ， 只要生产者数据准备好了就会发射出去 。 如果生产者比较慢， 则消费者会等待新的数据到来 。
 * 如果生产者比较快，则会有很 多数据发射给消费者 ， 而不管消费者当前有没有能力处理数据，这样就会导致背压 。 最后，在 RxJava 2.x
 * 中，只有新增的 Flowable 类型是支持背压的，并且 Flowable 很多操作 符内部都使用了背压策略，从而避免过多的数据填满内部的队列。
 * 
 * @author zhaosc
 *
 */
public class BackPressExample {

	private static Logger logger = Logger.getLogger(BackPressExample.class);


	@Test
	public void testBackPress() {
		Observable.create(new ObservableOnSubscribe<String>() {

			@Override
			public void subscribe(ObservableEmitter<String> emitter) throws Exception {
				for (int i = 0; i < 30000000; i++) {
					emitter.onNext("****Observable****" + i);
				}
			}
		}).subscribeOn(Schedulers.newThread())
		.observeOn(Schedulers.newThread()).subscribe((t) -> {
			SleepUtils.sleep(10000L);
			System.out.println(t);
		}, new Consumer<Throwable>() {

			@Override
			public void accept(Throwable t) throws Exception {
				t.printStackTrace();
			}
		});

		SleepUtils.sleep(600000000L);
	}

	@Test
	public void testFlowable() {
		Flowable.
		create(new FlowableOnSubscribe<String>() {

			@Override
			public void subscribe(FlowableEmitter<String> emitter) throws Exception {
				for (int i = 0; i < 30000000; i++) {
					emitter.onNext("****Flowable****" + i);
				}
			}
		}, BackpressureStrategy.MISSING)
		.subscribeOn(Schedulers.newThread()).observeOn(Schedulers.newThread())
				.subscribe((t) -> {
					SleepUtils.sleep(1000L);
					System.out.println(t);
				}, new Consumer<Throwable>() {

					@Override
					public void accept(Throwable t) throws Exception {
						t.printStackTrace();
					}
				});

		SleepUtils.sleep(600000L);
	}

}
